package edu.utexas.mapreduce;

import java.io.BufferedReader;
import java.io.FileReader;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.util.ArrayList;
import java.util.List;

import com.sun.net.httpserver.HttpServer;

import edu.utexas.ipc.IPC;

public class ServerMapReduce {

	private static ServerConfig config;
	private static List<InetSocketAddress> serverList;

	public ServerMapReduce (ServerConfig config){
	
		ServerMapReduce.config = config;
	}
	
	public void run() {
		try {

			int processID = config.processID;
			InetSocketAddress localAddress;
			
			if (config.serverList != null)
			{
				getServerList();
				config.HTTPport = serverList.get(processID).getPort() - 1000;
			}
			else {
				System.out.println("No server list specified.  Run in standalone mode");
				processID = 0;
				serverList = new ArrayList<InetSocketAddress>();
				serverList.add(new InetSocketAddress("localhost", 9000));
			}
			
			localAddress = serverList.get(processID);
			IPC ipc = new IPC(processID, localAddress, serverList);

			System.out.println("Starting web server @ " + config.HTTPport);
			System.out.println("Starting ipc server @ " + localAddress.getPort());
			System.out.println("  Base dir: " + config.baseDir);

			//
			// connect to other mapreduce servers
			//
			ipc.connect();
			
			//
			// Start web server to accept mapreduce jobs 
			//
			HttpServer server = HttpServer.create(new InetSocketAddress(config.HTTPport), 0);
			
			//SS: Create the shared job state object
			SharedMapState sState = new SharedMapState();
			
			// SS: pass shared state to handler
			server.createContext("/map_result", new HttpFileHandler("/map_result", config.baseDir, sState));
			server.createContext("/submit_job", new HttpSubmitJobHandler(config.baseDir, ipc, sState));
			server.createContext("/status", new HttpStatusRequestHandler(sState));
			server.createContext("/ping", new HttpPingHandler(sState));
			server.setExecutor(null); // creates a default executor
			server.start();
			
		} catch (IOException e) {
			e.printStackTrace();
		}
	}	
	public static List<InetSocketAddress> getServerList() throws IOException{
		if (serverList==null){
			serverList = new ArrayList<InetSocketAddress>();
			BufferedReader s = new BufferedReader(new FileReader(config.serverList));
			String address;
			
			while((address = s.readLine()) != null) {
				String[] temp = address.split(":");
				String host = temp[0];
				String port = temp[1];
				serverList.add(new InetSocketAddress(host, Integer.parseInt(port)));
			}
		}
		return serverList;
	}
	
	public static int  getPid(){
		return config.processID;
	}
	
}
